package com.ruyuan2020.im.c2g.service.impl;

import cn.hutool.core.date.StopWatch;
import com.ruyuan2020.im.common.core.domain.BasePage;
import com.ruyuan2020.im.common.hbase.HBaseTemplate;
import com.ruyuan2020.im.common.im.constant.Constants;
import com.ruyuan2020.im.common.im.domain.MessageJsonPush;
import com.ruyuan2020.im.common.im.domain.MessageSendJsonRequest;
import com.ruyuan2020.im.group.api.GroupApi;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author zhonghuashishan
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class MessageSendServiceImpl extends AbstractMessageSendService {

    private static final String GROUP_KEY_PREFIX = "GROUP";

    private static final int FACTOR = 128;

    private static final Integer MESSAGE_FETCH_THRESHOLD = 10;

    private final HBaseTemplate hbaseTemplate;

    @DubboReference(version = "1.0.0")
    private GroupApi groupApi;

    @Override
    public void save(List<MessageSendJsonRequest> requests, Long chatId) {
        StopWatch sw = new StopWatch("save");

        // 群聊消息，需要根据群聊id查询和获取群里的成员的id集合
        // 写扩散，读扩散，我都需要把这条消息扩散给这个群里的人都要知道
        // 但凡是限制一下群聊的人数，做写扩散都是可以的，效果也会好一些，简便一些
        // 微信朋友圈、微博，你的好友数量很多很多，你发一个朋友圈，发一条微博，少则需要几千人看到
        // 多则微博，几十万、几百万、几千万的粉丝关注了你，都要可以看到
        // 读扩散

        // 对于im模型，采取的还是读扩散，把群聊消息扩散给群里的其他人的时候，存储仅仅是一份
        // push或者pull模型，把群里最新的消息一定会推送给群聊里其他人的客户端

        // 通过redis里的list数据结构，lrange查询一个会话里的成员的信息
        // 如果说你要是拉了一个群以后，群主是你，群->chat，拉成员到群里来，群里是有成员，chat也会对应的是有成员
        // 但凡是拉到群里来的人，你是有对应的group群，你也有一个群chat，所以说这里是有这样的一套对应关系
        // 群&chat -> 一堆成员，可以在redis里是一个list数据结构
        // 每个成员 -> 一堆的chat列表（单聊chat，群聊chat），也可以是redis里的一个list数据结构

        // 账号、好友、群组、chat（单聊、群聊），java web服务，提供http接口
        // 数据存储可以是mysql+redis双写都可以，甚至说完全基于redis来存储，用list一类的数据结构来搞就可以了

        // db+redis双写的数据存储来查询群聊chat里的成员id集合
        sw.start("getMemberIdByGroupId");
        List<Long> memberIds = redisTemplate.opsForList().range(GROUP_KEY_PREFIX + "::" + chatId, 0, -1);
        if (CollectionUtils.isEmpty(memberIds)) {
            memberIds = groupApi.getMemberIdByGroupId(chatId);
        }
        sw.stop();

        // 将每一条群聊消息封装为hbase put请求
        List<Put> puts = new ArrayList<>(requests.size());
        for (MessageSendJsonRequest request : requests) {
            log.info("发送消息:{}", request.toJsonStr());

            // 根据群聊id+消息id，获取一个hbase rowkey，判断一下当前这个群聊消息是否已经存在了
            sw.start("exists");
            String rowKey = getRowKey(request.getChatId(), request.getMessageId());
            Get get = new Get(Bytes.toBytes(rowKey));
            boolean exists = hbaseTemplate.exists("c2g_message", get);
            sw.stop();

            if (!exists) {
                Put put = createMessagePut(rowKey, request);
                puts.add(put);
            }
        }

        // 将刚才封装好的每条群聊消息对应的一批hbase put请求，写入到hbase里去
        // 这里会把一批一批的群聊消息，每一批消息都是属于一个群聊的，每一次都是走一个batch模式写入到hbase里去
        // hbase是适合海量数据存储 ，高并发写入，底层海量数据是基于hdfs分布式文件存储
        sw.start("hbase.save");
        if (CollectionUtils.isNotEmpty(puts)) {
            this.hbaseTemplate.save("c2g_message", puts);
        }
        sw.stop();

        // 既然说群聊消息都已经落入到hbase存储里去
        // 必须把这些群聊的消息可以让群里的其他人可以看到
        Map<Long, List<MessageSendJsonRequest>> groupByToId = requests.stream().collect(Collectors.groupingBy(MessageSendJsonRequest::getToId));
        for (Map.Entry<Long, List<MessageSendJsonRequest>> entry : groupByToId.entrySet()) {
            // 如果说本次一个群聊里，新增的消息数量比较多，一个批次搞了超过10条
            // 就不走server端给客户端反向push的的动作
            // 我会通知让你的客户端来发送请求进行fetch，pull模型拉取消息
            if (entry.getValue().size() > MESSAGE_FETCH_THRESHOLD) {
                for(Long memberId : memberIds) {
                    sendMessageFetch(chatId, memberId);
                }
                for (MessageSendJsonRequest request : entry.getValue()) {
                    sendMessageSendResponse(request);
                }
            } else {
                for (MessageSendJsonRequest request : entry.getValue()) {
                    sendMessageSendResponse(request);
                    // 直接去进行push，server端主动push到客户端去
                    sendMessagePush(request, memberIds);
                }
            }
        }
        log.debug(sw.prettyPrint(TimeUnit.MILLISECONDS));
    }

    /**
     * 创建群聊消息put请求
     * @param rowKey
     * @param request
     * @return
     */
    private Put createMessagePut(String rowKey, MessageSendJsonRequest request) {
        // 群id的hash，
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(
                Bytes.toBytes("message_body"),
                Bytes.toBytes("fromId"),
                Bytes.toBytes(request.getFromId())
        );
        put.addColumn(
                Bytes.toBytes("message_body"),
                Bytes.toBytes("toId"),
                Bytes.toBytes(request.getToId())
        );
        put.addColumn(
                Bytes.toBytes("message_body"),
                Bytes.toBytes("messageId"),
                Bytes.toBytes(request.getMessageId())
        );
        put.addColumn(
                Bytes.toBytes("message_body"),
                Bytes.toBytes("messageType"),
                Bytes.toBytes(request.getMessageType())
        );
        put.addColumn(
                Bytes.toBytes("message_body"),
                Bytes.toBytes("content"),
                Bytes.toBytes(request.getContent())
        );
        put.addColumn(
                Bytes.toBytes("message_body"),
                Bytes.toBytes("sequence"),
                Bytes.toBytes(request.getSequence())
        );
        put.addColumn(
                Bytes.toBytes("message_body"),
                Bytes.toBytes("timestamp"),
                Bytes.toBytes(request.getTimestamp())
        );
        return put;
    }

    @Override
    public BasePage<MessageJsonPush> fetch(Long chatId, Long startMessageId, Long stopMessageId, int size) {
        // 当前对于客户端其实是知道每一个群聊里有哪些消息，拿到消息messageid最大的那一条
        // startMessageId，size
        Scan scan = new Scan();
        if (Objects.nonNull(startMessageId)) {
            String startRowKey = getRowKey(chatId, startMessageId);
            scan.withStartRow(Bytes.toBytes(startRowKey));
        }
        String stopRowKey = getRowKey(chatId, stopMessageId);
        scan.withStopRow(Bytes.toBytes(stopRowKey)).setLimit(size);

        List<MessageJsonPush> messageDTOS = hbaseTemplate.list("c2g_message", scan, (result) -> {
            MessageJsonPush push = new MessageJsonPush();
            push.setChatType(Constants.CHAT_TYPE_C2G);
            push.setFromId(Bytes.toLong(result.getValue(Bytes.toBytes("message_body"), Bytes.toBytes("fromId"))));
            push.setToId(Bytes.toLong(result.getValue(Bytes.toBytes("message_body"), Bytes.toBytes("toId"))));
            push.setMessageId(Bytes.toLong(result.getValue(Bytes.toBytes("message_body"), Bytes.toBytes("messageId"))));
            push.setMessageType(Bytes.toInt(result.getValue(Bytes.toBytes("message_body"), Bytes.toBytes("messageType"))));
            push.setContent(Bytes.toString(result.getValue(Bytes.toBytes("message_body"), Bytes.toBytes("content"))));
            push.setSequence(Bytes.toLong(result.getValue(Bytes.toBytes("message_body"), Bytes.toBytes("sequence"))));
            push.setTimestamp(Bytes.toLong(result.getValue(Bytes.toBytes("message_body"), Bytes.toBytes("timestamp"))));
            return push;
        });

        return new BasePage<>(messageDTOS);
    }

    private String getRowKey(Long groupId, Long messageId) {
        Long tempMessageId = messageId;
        if (Objects.isNull(tempMessageId)) {
            tempMessageId = 0L;
        }
        // 逆序消息id，做排序用，最新的消息-先查找
        // hbase里面会默认的按照rowkey进行升序排序，message id最新的消息反而越大，反而排的越靠后
        // 用max_value做一个减法，差值是越小，拼接在rowkey里来排序，群聊里越新的消息，rowkey是越小的
        // 查一个群聊的消息的时候，可以把越新的消息优先查询出来
        return getRowKey(groupId)
                + StringUtils.leftPad(String.valueOf(Long.MAX_VALUE - tempMessageId), 19, '0');
    }

    private String getRowKey(Long groupId) {
        int hash = (int) (groupId % FACTOR);
        // 3位群会话id的hash+群id+逆序消息id
        // hash，是为了在hbase中分区存储，相当于mysql的分表，把你的群聊消息，均匀的打散到hbase各个region里去
        // 为了避免说大量的群聊消息的读写集中在hbase某个table region里，数据热点问题
        // 对应rowkey都会取hash值，factor是table有多少个region，把你路由到某一个region里去
        // 就是说你路由到了哪个region里去，region序号
        // 群会话id，rowKey做群消息查询
        return StringUtils.leftPad(String.valueOf(hash), 3, '0') + "|"
                + StringUtils.leftPad(String.valueOf(groupId), 19, '0') + "|";
    }
}
